package vip.zhenzicheng.kafka.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import vip.zhenzicheng.kafka.properties.KafkaTopicProperties;

/**
 * 消费者
 *
 * @author zhenzicheng
 * @date 2022-09-27 17:59
 */
@Component
@Slf4j
public class MyConsumer {

  @Autowired
  private KafkaTopicProperties topicProperties;

  @Value("${spring.kafka.topics.my-test}")
  private String myTestTopic;

  /**
   * 单条普通消费，来一条消费一次
   * <pre>{@code
   * @KafkaListener(groupId = "testGroup", topicPartitions = {
   *             @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
   *             @TopicPartition(topic = "topic2", partitions = "0",
   *                     partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
   *     },concurrency = "6")
   * }</pre>
   * // concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
   *
   * @param record 消费记录，包含本次接收的消息
   * @param ack    手动消息确认
   */
  @KafkaListener(topics = "${spring.kafka.topics.my-test}", groupId = "${spring.kafka.consumer.group-id}")
  public void listenGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    // 本次监听到的kafka结果值
    String value = record.value();
    log.info("收到消息: {}", value);
    // 手动确认签收
    // 如果不签收也不提交offset，则每次项目启动都会重复获取到之前的消息，也即重复消费消息
    // ack.acknowledge();
  }

  /**
   * 配置多个消费组，模拟广播消息<br/>
   * 当多个消费组消费同一个topic就是广播方式
   *
   * @param record record
   * @param ack    ack
   */
  @KafkaListener(topics = "${spring.kafka.topics.my-test}", groupId = "custom_consumer_group_test_zzc_1")
  public void listenMultiGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    String value = record.value();
    log.info("收到消息: {}", value);
  }
}
